package com.atguigu.kafka.consumer.ConsumerGroup;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

/**
 * day11：
 *     Kafka_消费者_消费者组
 *
 *     测试：
 *          1、运行 CustomConsumer.java、CustomConsumer1.java、CustomConsumer2.java
 *          2、运行 CustomProducerCallback.java
 *          3、查看 CustomConsumer.java、CustomConsumer1.java、CustomConsumer2.java 控制台输出，有值代表进行了消费
 */

/**
 * day12：
 *     Kafka_消费者 Range/Roundrobin/Sticky 分配及再平衡
 *
 *     前提：
 *          将first主题分区数增加到7：
 *              kafka-topics.bat --bootstrap-server localhost:9092 --alter --topic first --partitions 7
 *     测试：
 *          0、设置分区分配策略
 *          1、运行 CustomConsumer.java、CustomConsumer1.java、CustomConsumer2.java
 *          2、运行 CustomProducerCallback.java
 *          3、查看 CustomConsumer.java、CustomConsumer1.java、CustomConsumer2.java 控制台输出，看看不同的消费者所消费的分区
 *          4、停止 CustomConsumer.java 消费者，然后再运行 CustomProducerCallback.java，查看消费分区情况
 *              ，45s后再查看 CustomConsumer1.java、CustomConsumer2.java 的消费情况（结果：存活的消费者会将 CustomConsumer.java 未消费的消息消费掉）
 *          5、再次运行 CustomProducerCallback.java，查看再平衡情况（即消费者会再重新分配分区）
 */
public class CustomConsumer {

    public static void main(String[] args) {
        // 0 配置
        Properties properties = new Properties();

        // 连接 bootstrap.servers
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test5");

        // 设置分区分配策略（org.apache.kafka.clients.consumer.RangeAssignor、org.apache.kafka.clients.consumer.RoundRobinAssignor、
        // org.apache.kafka.clients.consumer.StickyAssignor、org.apache.kafka.clients.consumer.CooperativeStickyAssignor）
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.StickyAssignor");

        // 1 创建一个消费者  "", "hello"
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 2 订阅主题 first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        // 3 消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

            kafkaConsumer.commitAsync();
        }
    }
}
